package com.tg.rxjava;

import org.drools.core.time.TimeUtils;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.util.Random;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * hystrix就是这样运作的
 */
public class RxJavaSimpleHystrixCommand {

    public RxJavaSimpleHystrixCommand(boolean beThread) {
        this.beThread = beThread;
    }

    /**
     * 线程池or信号量
     */
    private boolean beThread;

    /**
     * fallback方法
     *
     * @return
     */
    public Integer fallback() {
        return -1;
    }

    /**
     * 实际执行方法
     * 随机正确or异常
     *
     * @return
     */
    public Integer run() {
        try {
            log("实际执行run()挂起3秒");
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException ex) {
        } finally {
            Random random = new Random();
            int i = random.nextInt(4);
            if (i > 0) {
                return 99;
            }
            throw new RuntimeException("dd");

        }

    }

    /**
     * hystrixCommand提供外部执行
     *
     * @return
     * @throws Exception
     */
    public Integer execute() throws Exception {
        //【hystrix命令】发送者：执行熔断器、缓存、绑定熔断方法等操作
        Observable<Integer> hystrixSender = Observable.defer(
                new Func0<Observable<Integer>>() {
                    public Observable<Integer> call() {
                        log("执行熔断器、缓存、绑定熔断方法等操作");

                        //【实际业务】发送者，并包装到【hystrix命令】发送者中
                        Observable<Integer> bizSender = Observable.defer(new Func0<Observable<Integer>>() {
                            public Observable<Integer> call() {
                                log("实际执行run()");
                                return Observable.just(RxJavaSimpleHystrixCommand.this.run());
                            }
                        });

                        if (RxJavaSimpleHystrixCommand.this.beThread) {
                            //当采用线程池，异步执行实际业务run()方法
                            return bizSender.subscribeOn(Schedulers.io());
                        }
                        //当采用信号量，同步执行实际业务run()方法
                        return bizSender;
                    }
                }

        ).onErrorResumeNext(
                new Func1<Throwable, Observable<Integer>>() {
                    public Observable<Integer> call(Throwable t) {
                        log("异常跳fallback");
                        return Observable.just(RxJavaSimpleHystrixCommand.this.fallback());
                    }
                }
        );


        Future<Integer> future = hystrixSender.toBlocking().toFuture();
        log("开始阻塞等待结果");
        Integer a = future.get();
        log("已获得结果-" + a.toString());
        return a;

    }

    public static void main(String[] args) throws Exception {

        RxJavaSimpleHystrixCommand command = new RxJavaSimpleHystrixCommand(false);
        Integer target = command.execute();

        TimeUnit.SECONDS.sleep(10);
    }

    private void log(String msg) {
        System.out.println(Thread.currentThread().getName() + "-" + msg);
    }
}
